package kernel_db.parser;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import common.entities.abst.AbstPersistentEntity;
import common.entities.enums.SEX;
import common.entities.persistent.*;
import kernel_db.db.*;


public class ParserDB {
	private DB db;
	
	private String moviesFilePath = "";
	private String ratingsFilePath = "";
	private String countriesFilePath = "";
	private String genresFilePath = "";
	private String actorsFilePath = "";
	private String actressesFilePath = "";
	private String directorsFilePath = "";
	private String currentFile = "";
	
	private int maxBatch;	// how many inserts do we want in 1 batch
	private int fetchSize;	// how many rows do we want to fetch each time
	private final int topMovieVotes = 10000;
	private final float topMovieRank = 6;
	private List<Movie> movieList;
	private List<Actor> actorList;
	private List<Director> directorList;
	private List<Country> countryList;
	private List<Genre> genreList;
	private Map<String, Integer> moviesHash;	//key=makeKeyFromMovie(..), data=movie id in database
	private Map<String, Integer> actorsHash;	//key=makeKeyFromActor(..), data=actor id in database
	private Map<String, Integer> directorsHash;	//key=makeKeyFromDirector(..), data=director id in database
	private Map<String, Integer> countriesHash;	//key=country_name(lower case), data=country id in database
	private Map<String, Integer> genresHash;	//key=genre_name(lower case), data=genre id in database
	private Map<Integer, Integer> ratingsHash;	//key=movie_id, data=votes in database
	private Map<String, Boolean> movieConnectionHash;	//key=makeKeyFromMovieConnection(..), data=if it's in the DB
	private Actor previousActor;
	private Director previousDirector;
	private boolean readHashFromDB;
	private boolean readConnectionHashFromDB;
	String[] tablesNames = {"movie", "actor", "actor_movie", "country", "country_movie", 
			"director", "director_movie", "genre", "genre_movie",
			"movie_released_in_year_question", "actor_played_in_movie_question",
			"genre_of_movie_question", "movie_actor_in_question",
			"movie_directed_by_question"};
	String[] fileNames = {"movies.list", "actors.list", "actresses.list", "directors.list", "countries.list", "genres.list", "ratings.list"};
	String[] filePaths = new String[fileNames.length];
	
	//the fields below are for debugging
	//TODO isDebug should be false in normal mode, and true if we want debug printing
	public boolean isDebug = false;
	public int counter = 0;			//in XmaxBatch
	public int counterPrint = 100;	//in XmaxBatch
	public int counterBreak = 1000000;	//in XmaxBatch
	Date start = null, stop;
	Date start1 = null, stop1;
	Date start2 = null, stop2;
	
	
	public ParserDB(DB db, int maxBatch, int fetchSize) {
		this.db = db;
		this.maxBatch = maxBatch;
		this.fetchSize = fetchSize;
		
		movieList = new LinkedList<Movie>();
		actorList = new LinkedList<Actor>();
		directorList = new LinkedList<Director>();
		countryList = new LinkedList<Country>();
		genreList = new LinkedList<Genre>();
		
		moviesHash = new HashMap<String, Integer>();
		actorsHash = new HashMap<String, Integer>();
		directorsHash = new HashMap<String, Integer>();
		countriesHash = new HashMap<String, Integer>();
		genresHash = new HashMap<String, Integer>();
		ratingsHash = new HashMap<Integer, Integer>();
		
		movieConnectionHash = new HashMap<String, Boolean>();
		
		previousActor = null;
		previousDirector = null;
		readHashFromDB = false;
		readConnectionHashFromDB = false;
	}
	
	
	/**
	 * setting the maxBatch
	 * @param maxBatch size
	 */
	public void setMaxBatch(int maxBatch) {
		this.maxBatch = maxBatch;
	}
	/**
	 * setting the fetchSize
	 * @param fetchSize
	 */
	public void setFetchSize(int fetchSize) {
		this.fetchSize = fetchSize;
	}
	/**
	 * setting the movies file path
	 * @param movies DB file path (movies.list)
	 */
	public void setMoviesFilePath(String movieFilePath) {
		this.moviesFilePath = movieFilePath;
		filePaths[0] = movieFilePath;
	}
	/**
	 * setting the ratings file path
	 * @param ratings DB file path (ratings.list)
	 */
	public void setRatingsFilePath(String ratingsFilePath) {
		this.ratingsFilePath = ratingsFilePath;
		filePaths[6] = ratingsFilePath;
	}
	/**
	 * setting the countries file path
	 * @param countries DB file path (countries.list)
	 */
	public void setCountriesFilePath(String countryFilePath) {
		this.countriesFilePath = countryFilePath;
		filePaths[4] = countryFilePath;
	}
	/**
	 * setting the genres file path
	 * @param genres DB file path (genres.list)
	 */
	public void setGenresFilePath(String genreFilePath) {
		this.genresFilePath = genreFilePath;
		filePaths[5] = genreFilePath;
	}
	/**
	 * setting the actors file path
	 * @param actors DB file path (actors.list)
	 */
	public void setActorsFilePath(String actorFilePath) {
		this.actorsFilePath = actorFilePath;
		filePaths[1] = actorFilePath;
	}
	/**
	 * setting the actresses file path
	 * @param actresses DB file path (actresses.list)
	 */
	public void setActressesFilePath(String actressesFilePath) {
		this.actressesFilePath = actressesFilePath;
		filePaths[2] = actressesFilePath;
	}
	/**
	 * setting the directors file path
	 * @param directors DB file path (directors.list)
	 */
	public void setDirectorsFilePath(String directorFilePath) {
		this.directorsFilePath = directorFilePath;
		filePaths[3] = directorFilePath;
	}
	
	
	/**
	 * parsing the movies DB file and inserting data into tables
	 * @throws DB_Error if an error occurred
	 */
	private void parseMoviesFile() throws FileNotFoundException, DB_Error {
		// opening files
		FileInputStream fileStream = null;
		currentFile = moviesFilePath;
		fileStream = new FileInputStream(moviesFilePath);
		BufferedReader br = new BufferedReader(new InputStreamReader(fileStream));	
		
		// parsing file
		String line = "";
		Movie movie = new Movie();
		while (true && line != null) {
			try {
				line = br.readLine();
				if (Parser.parseMovieLine(line, movie)) {
					// handling Movie
					if (movie.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear()))
						continue;
					batchInsertMovie(movie, ParserAction.ParseMovie);
					
					movie = new Movie();
				}
			} catch (IOException e) { /* meaning: end of file */ break; }
		}
		
		// batch execute the remaining movies 
		batchExecuteInsertMovie();
		
		// closing files
		try {
			br.close();
			fileStream.close();
		} catch (IOException e) { }
	}
	/**
	 * parsing the actors\actresses DB file and inserting data into tables
	 * @throws DB_Error if an error occurred
	 * @param if maleActors is true then this function will parse the actors DB file,
	 * 		  if maleActors is false then it will parse the actresses DB file
	 */
	private void parseActorsFile(Boolean maleActors) throws FileNotFoundException, DB_Error {
		// opening files
		FileInputStream fileStream = null;
		if (maleActors) {
			currentFile = actorsFilePath;
			fileStream = new FileInputStream(actorsFilePath);
		}
		else {
			currentFile = actressesFilePath;
			fileStream = new FileInputStream(actressesFilePath);
		}
		BufferedReader br = new BufferedReader(new InputStreamReader(fileStream));	
		
		// parsing file
		parserReturnType returnType;
		String line = "";
		Movie movie = new Movie();
		Actor actor = new Actor();
		Actor newActor = new Actor();
		while (true && line != null) {
			try {
				line = br.readLine();
				
				returnType = Parser.parseActorLine(line, movie, newActor);
				if (returnType == parserReturnType.ParsedName) {
					if (maleActors)
						newActor.setSex(SEX.MALE);
					else
						newActor.setSex(SEX.FEMALE);
					
					actor = newActor;
					newActor = new Actor();
					
					returnType = parserReturnType.ParsedMovie;
				}
				if (returnType == parserReturnType.ParsedMovie) {
					// handling Actor-Movie
					if (actor.isViolateRestrictions() || movie.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear()))
						continue;
					if (maleActors)
						batchInsertMovieConnection(actor, movie, ParserAction.ParseActorMovie);
					else
						batchInsertMovieConnection(actor, movie, ParserAction.ParseActressesMovie);
					
					movie = new Movie();
				}
			} catch (IOException e) { /* meaning: end of file */ break; }
		}
		
		// batch execute the remaining actors-movies 
		batchExecuteInsertActorMovie();
		
		// closing files
		try {
			br.close();
			fileStream.close();
		} catch (IOException e) { }
	}
	/**
	 * parsing the directors DB file and inserting data into tables
	 * @throws DB_Error if an error occurred
	 */
	private void parseDirectorsFile() throws FileNotFoundException, DB_Error {
		// opening files
		FileInputStream fileStream = null;
		currentFile = directorsFilePath;
		fileStream = new FileInputStream(directorsFilePath);
		BufferedReader br = new BufferedReader(new InputStreamReader(fileStream));	
		
		// parsing file
		parserReturnType returnType;
		String line = "";
		Movie movie = new Movie();
		Director director = new Director();
		Director newDirector = new Director();
		while (true && line != null) {
			try {
				line = br.readLine();
				
				returnType = Parser.parseDirectorLine(line, movie, newDirector);
				if (returnType == parserReturnType.ParsedName) {
					director = newDirector;
					newDirector = new Director();
					
					returnType = parserReturnType.ParsedMovie;
				}
				if (returnType == parserReturnType.ParsedMovie) {
					// handling Director-Movie
					if (director.isViolateRestrictions() || movie.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear()))
						continue;
					batchInsertMovieConnection(director, movie, ParserAction.ParseDirectorMovie);
					
					movie = new Movie();
				}
			} catch (IOException e) { /* meaning: end of file */ break; }
		}
		
		// batch execute the remaining directors-movies 
		batchExecuteInsertDirectorMovie();
		
		// closing files
		try {
			br.close();
			fileStream.close();
		} catch (IOException e) { }
	}
	/**
	 * parsing the countries DB file and inserting data into tables
	 * @throws DB_Error if an error occurred
	 */
	private void parseCountriesFile() throws FileNotFoundException, DB_Error {
		// opening files
		FileInputStream fileStream = null;
		currentFile = countriesFilePath;
		fileStream = new FileInputStream(countriesFilePath);
		BufferedReader br = new BufferedReader(new InputStreamReader(fileStream));	
		
		// parsing file
		String line = "";
		Movie movie = new Movie();
		Country country = new Country();
		while (true && line != null) {
			try {
				line = br.readLine();
				if (Parser.parseCountryLine(line, movie, country)) {
					// handling Country-Movie
					if (country.isViolateRestrictions() || movie.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear()))
						continue;
					batchInsertMovieConnection(country, movie, ParserAction.ParseCountryMovie);
					
					movie = new Movie();
					country = new Country();
				}
			} catch (IOException e) { /* meaning: end of file */ break; }
		}
		
		// batch execute the remaining countries-movies 
		batchExecuteInsertCountryMovie();
		
		// closing files
		try {
			br.close();
			fileStream.close();
		} catch (IOException e) { }
	}
	/**
	 * parsing the genres DB file and inserting data into tables
	 * @throws DB_Error if an error occurred
	 */
	private void parseGenresFile() throws FileNotFoundException, DB_Error {
		// opening files
		FileInputStream fileStream = null;
		currentFile = genresFilePath;
		fileStream = new FileInputStream(genresFilePath);
		BufferedReader br = new BufferedReader(new InputStreamReader(fileStream));	
		
		// parsing file
		String line = "";
		Movie movie = new Movie();
		Genre genre = new Genre();
		while (true && line != null) {
			try {
				line = br.readLine();
				if (Parser.parseGenreLine(line, movie, genre)) {
					// handling Genre-Movie
					if (genre.isViolateRestrictions() || movie.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear()))
						continue;
					batchInsertMovieConnection(genre, movie, ParserAction.ParseGenreMovie);
					
					movie = new Movie();
					genre = new Genre();
				}
			} catch (IOException e) { /* meaning: end of file */ break; }
		}
		
		// batch execute the remaining genres-movies 
		batchExecuteInsertGenreMovie();
		
		// closing files
		try {
			br.close();
			fileStream.close();
		} catch (IOException e) { }
	}
	/**
	 * parsing the ratings DB file and inserting data into tables
	 * @throws DB_Error if an error occurred
	 */
	private void parseRatingsFile() throws FileNotFoundException, DB_Error {
		// opening files
		FileInputStream fileStream = null;
		currentFile = ratingsFilePath;
		fileStream = new FileInputStream(ratingsFilePath);
		BufferedReader br = new BufferedReader(new InputStreamReader(fileStream));	
		
		// parsing file
		String line = "";
		Movie movie = new Movie();
		while (true && line != null) {
			try {
				line = br.readLine();
				if (Parser.parseRatingLine(line, movie)) {
					// handling Rating of a Movie
					if (movie.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear()))
						continue;
					batchInsertMovie(movie, ParserAction.ParseRatingsMovie);
					
					movie = new Movie();
				}
			} catch (IOException e) { /* meaning: end of file */ break; }
		}
		
		// batch execute the remaining ratings of movies 
		batchExecuteInsertRatingMovie();
		
		// closing files
		try {
			br.close();
			fileStream.close();
		} catch (IOException e) { }
	}

	
	/**
	 * parse all files. this is the function the kernel is calling for import/update the DB.
	 * @throws DB_Error if an sql error occurred
	 */
	public boolean parseAllFiles() throws DB_Error {
		checkFilesPaths();
		
		for (int i=0; i <= 7; i++) {
			try {
				switch(i) {
				case 0:
					debugPrint("Parser maxBatch=" + maxBatch + ", counterPrint=" + counterPrint + ", counterBreak=" + counterBreak);
					debugPrint("");
					calculateElapsedTime(true, "Parsing Movie File");
					parseMoviesFile();
					break;
				case 1:
					calculateElapsedTime(false, "Parsing Movie File");
					calculateElapsedTime(true, "Parsing Actor File");
					readHashFromDB = false;
					readConnectionHashFromDB = false;
					parseActorsFile(true);
					actorsHash.clear();
					break;
				case 2:
					calculateElapsedTime(false, "Parsing Actor File");
					calculateElapsedTime(true, "Parsing Actresses File");
					readHashFromDB = false;
					parseActorsFile(false);
					actorsHash.clear();
					movieConnectionHash.clear();
					break;
				case 3:
					calculateElapsedTime(false, "Parsing Actresses File");
					calculateElapsedTime(true, "Parsing Director File");
					readHashFromDB = false;
					readConnectionHashFromDB = false;
					parseDirectorsFile();
					directorsHash.clear();
					movieConnectionHash.clear();
					break;
				case 4:
					calculateElapsedTime(false, "Parsing Director File");
					calculateElapsedTime(true, "Parsing Country File");
					readHashFromDB = false;
					readConnectionHashFromDB = false;
					parseCountriesFile();
					countriesHash.clear();
					movieConnectionHash.clear();
					break;
				case 5:
					calculateElapsedTime(false, "Parsing Country File");
					calculateElapsedTime(true, "Parsing Genre File");
					readHashFromDB = false;
					readConnectionHashFromDB = false;
					parseGenresFile();
					genresHash.clear();
					movieConnectionHash.clear();
					break;
				case 6:
					calculateElapsedTime(false, "Parsing Genre File");
					calculateElapsedTime(true, "Parsing Rating File");
					readHashFromDB = false;
					parseRatingsFile();
					moviesHash.clear();
					ratingsHash.clear();
					break;
				case 7:
					calculateElapsedTime(false, "Parsing Rating File");
					debugPrint("");
					clearAllParams();
					break;
				default:
					break;
				}
			} catch (FileNotFoundException e) {
				clearAllParams();
				throw new DB_Error("File '" + currentFile + "' not found.", false);
//			} catch (DB_Error db_e) {
//				//TODO disable this block in normal mode, and enable it if we want to debug
//				if (i != 1)
//					movieConnectionHash.clear();
//				actorsHash.clear();
//				directorsHash.clear();
//				countriesHash.clear();
//				genresHash.clear();
//				if (!db_e.toString().equals("No error message specified"))
//					db_e.printStackTrace();
			}
		}

		clearAllParams();
		
		return true;
	}
	/**
	 * deletes all the data in the DB
	 * @throws DB_Error if the delete has failed
	 */
	public void deleteAll() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			PreparedStatement stmt = null;
			for (String tableName : tablesNames) {
				stmt = conn.prepareStatement("DELETE FROM " + tableName);
				stmt.executeUpdate();
			}
			stmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			throw new DB_Error("ERROR deleting all - " + e.getMessage());
		}
		
		db.enqueueConnection(conn);
	}
	/**
	 * @throws DB_Error if one of the file paths is illegal
	 */
	public void checkFilesPaths() throws DB_Error {
		for (int i = 0; i < filePaths.length; i++) {
			if (filePaths[i] == null)
				throw new DB_Error(fileNames[i] + " file path is illegal. Please check that the path in the .ini file is valid.", false);
		}
		
		try {
			for (int i = 0; i < filePaths.length; i++) {
				try {
					FileInputStream fileStream = new FileInputStream(filePaths[i]);
					fileStream.close();
				} catch (FileNotFoundException e) {
					throw new DB_Error(fileNames[i] + " file path '" + filePaths[i] + "' wasn't found. Please check that the path in the .ini file is valid.", false);
				}
			}
		} catch (IOException e) { /*do nothing*/ }
		
		return;
	}
	

	
	/**
	 * dropping the dropType(FOREIGN KEY or INDEX) from tableName for foreignKeys
	 * @param conn - an open connection to DB
	 * @param tableName - a table name
	 * @param dropType - can be "FOREIGN KEY" or "INDEX"
	 * @param foreignKeys - an array of foreignKeys we want to drop
	 * @throws DB_Error if drop failed
	 */
	private void dropFromTable(Connection conn, String tableName, String dropType, String[] foreignKeys) throws DB_Error {
		try {
			Statement stmt = conn.createStatement();
			
			String sqlString = "ALTER TABLE " + tableName;
			int index = 0;
			for (String fk : foreignKeys) {
				if (index == 0)
					sqlString = sqlString.concat(" DROP " + dropType + " " + fk);
				else
					sqlString = sqlString.concat(" , DROP " + dropType + " " + fk);
				
				index++;
			}
			stmt.execute(sqlString);
			// closing the Statement
			stmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			throw new DB_Error("ERROR droping the table " + tableName + " foreign keys - " + e.getMessage());
		}
	}
	/**
	 * altering all the tables in the DB to InnoDB or to MyISAM.
	 * @param dbType - can be InnoDB or MyISAM
	 * @param conn - an open connection to DB
	 * @throws DB_Error if alter tables failed
	 */
	private void alterTablesTo(DBType dbType, Connection conn) throws DB_Error {
		try {
			Statement stmt = conn.createStatement();
			
			for (String table : tablesNames) {
				stmt.execute("ALTER TABLE " + table + " ENGINE = " + dbType.getName());
			}
			// closing the Statement
			stmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			throw new DB_Error("ERROR altering the DB tables to " + dbType.getName() + " - " + e.getMessage());
		}
	}
	/**
	 * changing the DB to MyISAM
	 * @throws DB_Error if change failed
	 */
	public void changeTablesToMyISAM() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			Statement stmt = conn.createStatement();
			
			dropFromTable(conn, "actor_movie", "FOREIGN KEY", new String[] {"movieTOactor", "actorTOmovie"});
			dropFromTable(conn, "director_movie", "FOREIGN KEY", new String[] {"movieTOdirector", "directorTOmovie"});
			dropFromTable(conn, "country_movie", "FOREIGN KEY", new String[] {"movieTOcountry", "countryTOmovie"});
			dropFromTable(conn, "genre_movie", "FOREIGN KEY", new String[] {"movieTOgenre", "genreTOmovie"});
		
			
			dropFromTable(conn, "actor_played_in_movie_question", "FOREIGN KEY", new String[] {"movieAPIMQ", "APIMQw3", "APIMQw2", "APIMQw1", "APIMQr"});
			
			dropFromTable(conn, "movie_released_in_year_question", "FOREIGN KEY", new String[] {"MRIYQw3", "MRIYQw2", "MRIYQw1", "MRIYQr"});
			
			dropFromTable(conn, "genre_of_movie_question", "FOREIGN KEY", new String[] {"movieGOMQ", "GOMQw3", "GOMQw2", "GOMQw1", "GOMQr"});

			dropFromTable(conn, "movie_actor_in_question", "FOREIGN KEY", new String[] {"actorMANIQ", "MANIQw3", "MANIQw2", "MANIQw1", "MANIQr"});
			
			dropFromTable(conn, "movie_directed_by_question", "FOREIGN KEY", new String[] {"directorMDBQ", "MDBQw3", "MDBQw2", "MDBQw1", "MDBQr"});
			
			// closing the Statement
			stmt.close();
			
			alterTablesTo(DBType.MyISAM, conn);
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			throw new DB_Error("ERROR changing the DB tables to " + DBType.MyISAM.getName() + " - " + e.getMessage());
		}
		
		db.enqueueConnection(conn);
	}
	/**
	 * changing the DB to InnoDB
	 * @throws DB_Error if change failed
	 */
	public void changeTablesToInnoDB() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			alterTablesTo(DBType.InnoDB, conn);

			Statement stmt = conn.createStatement();
			
			stmt.execute("ALTER TABLE actor_movie " + 
							"ADD CONSTRAINT actorTOmovie  FOREIGN KEY (actor_id)  REFERENCES actor (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE,  " +
							"ADD CONSTRAINT movieTOactor  FOREIGN KEY (movie_id)  REFERENCES movie (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE");
			stmt.execute("ALTER TABLE director_movie " + 
							"ADD CONSTRAINT directorTOmovie  FOREIGN KEY (director_id)  REFERENCES director (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE,  " +
							"ADD CONSTRAINT movieTOdirector  FOREIGN KEY (movie_id)  REFERENCES movie (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE");
			stmt.execute("ALTER TABLE country_movie " + 
							"ADD CONSTRAINT countryTOmovie  FOREIGN KEY (country_id)  REFERENCES country (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE,  " +
							"ADD CONSTRAINT movieTOcountry  FOREIGN KEY (movie_id)  REFERENCES movie (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE");
			stmt.execute("ALTER TABLE genre_movie " + 
							"ADD CONSTRAINT genreTOmovie  FOREIGN KEY (genre_id)  REFERENCES genre (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE,  " +
							"ADD CONSTRAINT movieTOgenre  FOREIGN KEY (movie_id)  REFERENCES movie (id) " +
								"ON DELETE CASCADE  ON UPDATE CASCADE");
			
			
			stmt.execute("ALTER TABLE actor_played_in_movie_question " +  
							  "ADD CONSTRAINT APIMQr  FOREIGN KEY (ans_actor_r)  REFERENCES actor (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE, " +  
							  "ADD CONSTRAINT APIMQw1  FOREIGN KEY (ans_actor_w1)  REFERENCES actor (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT APIMQw2  FOREIGN KEY (ans_actor_w2)  REFERENCES actor (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT APIMQw3  FOREIGN KEY (ans_actor_w3)  REFERENCES actor (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT movieAPIMQ  FOREIGN KEY (movie_id)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE ");
			
			stmt.execute("ALTER TABLE movie_released_in_year_question " +
							 "ADD CONSTRAINT MRIYQr  FOREIGN KEY (ans_movie_r)  REFERENCES movie (id) " +
							 	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +
							 "ADD CONSTRAINT MRIYQw1  FOREIGN KEY (ans_movie_w1)  REFERENCES movie (id) " +
							 	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +
							 "ADD CONSTRAINT MRIYQw2  FOREIGN KEY (ans_movie_w2)  REFERENCES movie (id) " +
							 	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +
							 "ADD CONSTRAINT MRIYQw3  FOREIGN KEY (ans_movie_w3)  REFERENCES movie (id) " +
							 	"ON DELETE CASCADE  ON UPDATE CASCADE ");
			
			stmt.execute("ALTER TABLE genre_of_movie_question " +  
							  "ADD CONSTRAINT GOMQr  FOREIGN KEY (ans_genre_r)  REFERENCES genre (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE, " +  
							  "ADD CONSTRAINT GOMQw1  FOREIGN KEY (ans_genre_w1)  REFERENCES genre (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT GOMQw2  FOREIGN KEY (ans_genre_w2)  REFERENCES genre (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT GOMQw3  FOREIGN KEY (ans_genre_w3)  REFERENCES genre (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT movieGOMQ  FOREIGN KEY (movie_id)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE ");
			
			stmt.execute("ALTER TABLE movie_actor_in_question " +  
							  "ADD CONSTRAINT MANIQr  FOREIGN KEY (ans_movie_r)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE, " +  
							  "ADD CONSTRAINT MANIQw1  FOREIGN KEY (ans_movie_w1)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT MANIQw2  FOREIGN KEY (ans_movie_w2)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT MANIQw3  FOREIGN KEY (ans_movie_w3)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT actorMANIQ  FOREIGN KEY (actor_id)  REFERENCES actor (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE ");
			
			stmt.execute("ALTER TABLE movie_directed_by_question " +  
							  "ADD CONSTRAINT MDBQr  FOREIGN KEY (ans_movie_r)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE, " +  
							  "ADD CONSTRAINT MDBQw1  FOREIGN KEY (ans_movie_w1)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT MDBQw2  FOREIGN KEY (ans_movie_w2)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT MDBQw3  FOREIGN KEY (ans_movie_w3)  REFERENCES movie (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE,  " +  
							  "ADD CONSTRAINT directorMDBQ  FOREIGN KEY (director_id)  REFERENCES director (id) " +  
							  	"ON DELETE CASCADE  ON UPDATE CASCADE ");
			
			
			// closing the Statement
			stmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			throw new DB_Error("ERROR changing the DB tables to " + DBType.InnoDB.getName() + " - " + e.getMessage());
		}
		
		db.enqueueConnection(conn);
	}
	
	
	/**
	 * getting the ids and setting them for all movies in movieList
	 * @throws SQLException if query failed
	 */
	private synchronized void getMoviesIds() throws SQLException {
		Integer movieId;
		for (Movie m : movieList) {
			//check if movie exist, and if yes get his id
			movieId = moviesHash.get(makeKeyFromMovie(m));
			if (movieId != null)
				m.setId(movieId);
		}
	}
	/**
	 * returns the key for the moviesHash
	 * @param a movie we want to calculate a key for
	 */
	private String makeKeyFromMovie(Movie movie) {
		return movie.getTitle().toLowerCase() + movie.getReleaseYear(); 
	}
	/**
	 * returns the key for the actorsHash
	 * @param an actor we want to calculate a key for
	 */
	private String makeKeyFromActor(Actor actor) {
		return actor.getFirstName().toLowerCase() + actor.getLastNameWithPersonNumber().toLowerCase() + actor.getSex().getValue(); 
	}	
	/**
	 * returns the key for the directorsHash
	 * @param a director we want to calculate a key for
	 */
	private String makeKeyFromDirector(Director director) {
		return director.getFirstName().toLowerCase() + director.getLastNameWithPersonNumber().toLowerCase(); 
	}
	/**
	 * returns the key for the movieConnectionHash
	 * @param a entity id and a movie id we want to calculate a key for
	 */
	private String makeKeyFromMovieConnection(int entityID, int movieID) {
		return entityID + "." + movieID; 
	}
	
	
	/**
	 * inserts movie to batch. this function is for import and update as well.
	 * @param action can be ParseMovie\ParseRatingsMovie
	 * @throws DB_Error if batch insert failed
	 */
	public synchronized void batchInsertMovie(Movie movie, ParserAction action) throws DB_Error {
		buildHashFromDB(action);
		movieList.add(movie);
		
		if (isDebug && counter == 0 && movieList.size() == 1)
			calculateElapsedTime(true);
		
		if (movieList.size() < maxBatch)
			return;
		else {
			switch (action) {
				case ParseMovie:
					batchExecuteInsertMovie();
					break;
				case ParseRatingsMovie:
					batchExecuteInsertRatingMovie();
					break;
			}
			
			if (isDebug) {
				DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
				Date date = new Date();
				if (counter % counterPrint == 0) {
					System.out.println("\t\t " + action.getText() + " Parser counter:" + counter + ", at time " + dateFormat.format(date) + ", Time Elapsed=" + calculateElapsedTime(false));
					calculateElapsedTime(true);
				}			
				counter++;
				if (counter % counterBreak == 0) 
					throw new DB_Error();
			}
		}
	}
	/**
	 * inserts <actor\director\genre\country>-movie to batch. this function is for import and update as well.
	 * @param action can be ParseActorMovie\ParseDirectorMovie\ParseCountryMovie\ParseGenreMovie
	 * @throws DB_Error if batch insert failed
	 */
	public synchronized void batchInsertMovieConnection(AbstPersistentEntity entity, Movie movie, ParserAction action) throws DB_Error {
		switch (action) {
			case ParseActorMovie:
			case ParseActressesMovie:
				Actor actor = (Actor) entity;
				actorList.add(actor);
				break;
			case ParseDirectorMovie:
				Director director = (Director) entity;
				directorList.add(director);
				break;
			case ParseCountryMovie:
				Country country = (Country) entity;
				countryList.add(country);
				break;
			case ParseGenreMovie:
				Genre genre = (Genre) entity;
				genreList.add(genre);
				break;
		}
		buildHashFromDB(action);
		buildMovieConnectionHashFromDB(action);
		movieList.add(movie);

		if (isDebug && counter == 0 && movieList.size() == 1)
			calculateElapsedTime(true);
		
		if (movieList.size() < maxBatch)
			return;
		else {
			switch (action) {
				case ParseActorMovie:
				case ParseActressesMovie:
					batchExecuteInsertActorMovie();
					break;
				case ParseDirectorMovie:
					batchExecuteInsertDirectorMovie();
					break;
				case ParseCountryMovie:
					batchExecuteInsertCountryMovie();
					break;
				case ParseGenreMovie:
					batchExecuteInsertGenreMovie();
					break;
			}
			
			if (isDebug) {
				DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
				Date date = new Date();
				if (counter % counterPrint == 0) {
					debugPrint("\t\t " + action.getText() + " Parser counter:" + counter + ", at time " + dateFormat.format(date) + ", Time Elapsed=" + calculateElapsedTime(false));
					calculateElapsedTime(true);
				}	
				counter++;
				if (counter % counterBreak == 0)
					throw new DB_Error();
			}
			
		}
	}
	/**
	 * builds an hash table from the data in the DB. this is used when we update the DB
	 * @param action can be ParseMovie\ParseActorMovie\ParseActressesMovie\ParseDirectorMovie\ParseCountryMovie\ParseGenreMovie
	 * @throws DB_Error if a sql query failed
	 */
	private synchronized void buildHashFromDB(ParserAction action) throws DB_Error {
		if (readHashFromDB)
			return;
		readHashFromDB = true;
		
		Map<String, Integer> hash = null;
		Map<Integer, Integer> rHash = null;
		String sqlQuery = "";
		
		switch (action) {
			case ParseMovie:
				hash = moviesHash;
				sqlQuery = "SELECT id, IMDB_name, release_year FROM movie WHERE IMDB_name IS NOT NULL";
				break;
			case ParseActorMovie:
				hash = actorsHash;
				sqlQuery = "SELECT id, IMDB_first_name, IMDB_last_name FROM actor WHERE Sex=" + SEX.MALE.getValue() + " AND IMDB_first_name IS NOT NULL";
				break;
			case ParseActressesMovie:
				hash = actorsHash;
				sqlQuery = "SELECT id, IMDB_first_name, IMDB_last_name FROM actor WHERE Sex=" + SEX.FEMALE.getValue() + " AND IMDB_first_name IS NOT NULL";
				break;
			case ParseDirectorMovie:
				hash = directorsHash;
				sqlQuery = "SELECT id, IMDB_first_name, IMDB_last_name FROM director WHERE IMDB_first_name IS NOT NULL";
				break;
			case ParseCountryMovie:
				hash = countriesHash;
				sqlQuery = "SELECT DISTINCT id, IMDB_name FROM country WHERE IMDB_name IS NOT NULL";
				break;
			case ParseGenreMovie:
				hash = genresHash;
				sqlQuery = "SELECT DISTINCT id, IMDB_name FROM genre WHERE IMDB_name IS NOT NULL";
				break;
			case ParseRatingsMovie:
				rHash = ratingsHash;
				sqlQuery = "SELECT id, IMDB_name, votes FROM movie WHERE IMDB_name IS NOT NULL";
				break;
		}
		if (hash != null && hash.size() != 0)
			return;
		if (rHash != null && rHash.size() != 0)
			return;
		
		Connection conn=db.dequeueConnection();
		String key = null;
		Movie movie = new Movie();
		Actor actor = new Actor();
		Director director = new Director();
		int id = 0, votes = 0;
		
		try {
			Statement stmt = conn.createStatement();
			stmt.setFetchSize(fetchSize);
			ResultSet rs;
			rs = stmt.executeQuery(sqlQuery);
		
			while (rs.next()) {
				switch (action) {
					case ParseMovie:
						movie.setTitle(rs.getString("IMDB_name"));
						movie.setReleaseYear(rs.getInt("release_year"));
						key = makeKeyFromMovie(movie);
						break;
					case ParseActorMovie:
						actor.setFirstName(rs.getString("IMDB_first_name"));
						actor.setLastName(rs.getString("IMDB_last_name"));
						actor.setSex(SEX.MALE);
						key = makeKeyFromActor(actor);
						break;
					case ParseActressesMovie:
						actor.setFirstName(rs.getString("IMDB_first_name"));
						actor.setLastName(rs.getString("IMDB_last_name"));
						actor.setSex(SEX.FEMALE);
						key = makeKeyFromActor(actor);
						break;
					case ParseDirectorMovie:
						director.setFirstName(rs.getString("IMDB_first_name"));
						director.setLastName(rs.getString("IMDB_last_name"));
						key = makeKeyFromDirector(director);
						break;
					case ParseCountryMovie:
					case ParseGenreMovie:
						key = rs.getString("IMDB_name").toLowerCase();
						break;
					case ParseRatingsMovie:
						id = rs.getInt("id");
						votes = rs.getInt("votes");
						break;
				}
				
				// inserting key+id to hash
				if (hash != null)
					hash.put(key, rs.getInt("id"));
				else
					rHash.put(id, votes);
					
			}
			
			rs.close();
			stmt.close();
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			throw new DB_Error("ERROR building hash - " + e.getMessage());
		}
		db.enqueueConnection(conn);
	}
	/**
	 * builds an connection hash from movies to <action> table from the data in the DB. this is used when we update the DB
	 * @param action can be ParseMovie\ParseActorMovie\ParseActressesMovie\ParseDirectorMovie\ParseCountryMovie\ParseGenreMovie
	 * @throws DB_Error if a sql query failed
	 */
	private synchronized void buildMovieConnectionHashFromDB(ParserAction action) throws DB_Error {
		if (readConnectionHashFromDB)
			return;
		readConnectionHashFromDB = true;
		
		Map<String, Boolean> hash = null;
		String sqlQuery = "";

		hash = movieConnectionHash;
		switch (action) {
			case ParseActorMovie:
				sqlQuery = "SELECT actor_id, movie_id FROM actor_movie";
				break;
			case ParseActressesMovie:
				sqlQuery = "SELECT actor_id, movie_id FROM actor_movie";
				break;
			case ParseDirectorMovie:
				sqlQuery = "SELECT director_id, movie_id FROM director_movie";
				break;
			case ParseCountryMovie:
				sqlQuery = "SELECT country_id, movie_id FROM country_movie";
				break;
			case ParseGenreMovie:
				sqlQuery = "SELECT genre_id, movie_id FROM genre_movie";
				break;
		}
		if (hash.size() != 0)
			return;
		
		Connection conn=db.dequeueConnection();
		String key = null;
		
		try {
			Statement stmt = conn.createStatement();
			stmt.setFetchSize(fetchSize);
			ResultSet rs;
			rs = stmt.executeQuery(sqlQuery);
		
			while (rs.next()) {
				switch (action) {
					case ParseActorMovie:
					case ParseActressesMovie:
					case ParseDirectorMovie:
					case ParseCountryMovie:
					case ParseGenreMovie:
						key = makeKeyFromMovieConnection(rs.getInt(1), rs.getInt(2));
						break;
				}
				
				// inserting key+data to hash
				hash.put(key, true);
			}
			
			rs.close();
			stmt.close();
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			throw new DB_Error("ERROR building hash - " + e.getMessage());
		}
		db.enqueueConnection(conn);
	}
	
	
	/**
	 * executing the batch insert/update movie.
	 * @throws DB_Error if batch insert/update failed
	 */
	public void batchExecuteInsertMovie() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			ResultSet rs;
			Statement stmt = conn.createStatement();
			PreparedStatement pstmt = conn.prepareStatement(
					"INSERT INTO movie(id, IMDB_name, release_year, name, rank, votes) VALUES(?, ?, ?, ?, ?, ?)");
			
			// calculating the current movie id
			int currentMovieId;
			rs = stmt.executeQuery("SELECT MAX(id) FROM movie");
			if (rs.next()) 
				currentMovieId = rs.getInt("MAX(id)") + 1;
			else
				currentMovieId = 1;
			// closing the Statement
			stmt.close();
			
			// inserting movies
			for (Movie m : movieList) {
				if (m.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear()))
					continue;
				//check if movie already exist
				if (moviesHash.containsKey(makeKeyFromMovie(m)))
					continue;
				
				//insert new movie to insert batch
				pstmt.setInt(1, currentMovieId);
				pstmt.setString(2, m.getTitle());	
				pstmt.setInt(3, m.getReleaseYear());
				pstmt.setString(4, m.getTitle());	
				pstmt.setFloat(5, m.getRank());
				pstmt.setInt(6, m.getVotes());
				pstmt.addBatch();
				
				//insert movie_key+id to hash
				moviesHash.put(makeKeyFromMovie(m), currentMovieId);
				currentMovieId++;
			}
			// executing the movie batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			clearAllParams();
			throw new DB_Error("ERROR inserting batch movie - " + e.getMessage());
		}
		
		// removing all movies from list
		movieList.clear();
		
		db.enqueueConnection(conn);
	}

	/**
	 * executing the batch insert/update actor-movie
	 * @throws DB_Error if batch insert failed
	 */
	public void batchExecuteInsertActorMovie() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			ResultSet rs;
			PreparedStatement pstmt;
			Statement stmt = conn.createStatement();
			pstmt = conn.prepareStatement(
					"INSERT INTO actor(id, IMDB_first_name, IMDB_last_name, Sex, first_name, last_name) VALUES(?, ?, ?, ?, ?, ?)");
			
			// calculating the current actor id
			int currentActorId;
			rs = stmt.executeQuery("SELECT MAX(id) FROM actor");
			if (rs.next()) 
				currentActorId = rs.getInt("MAX(id)") + 1;
			else
				currentActorId = 1;
			// closing the Statement
			stmt.close();
			
			// inserting actors
			for (Actor a : actorList) {
				if (a.isViolateRestrictions())
					continue;
				if (a == previousActor)
					continue;
				//check if actor already exist
				if (actorsHash.containsKey(makeKeyFromActor(a))) {
					a.setId(actorsHash.get(makeKeyFromActor(a)));
					continue;
				}
				
				//insert new actor to batch
				pstmt.setInt(1, currentActorId);
				pstmt.setString(2, a.getFirstName());	
				pstmt.setString(3, a.getLastNameWithPersonNumber());
				pstmt.setInt(4, a.getSex().getValue());
				pstmt.setString(5, a.getFirstName());	
				pstmt.setString(6, a.getLastName());	
				pstmt.addBatch();
				
				previousActor = a;
				a.setId(currentActorId);
				//insert actor_key+id to hash
				actorsHash.put(makeKeyFromActor(a), currentActorId);
				currentActorId++;
			}
			// executing the actor batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
			// getting movies ids
			getMoviesIds();
			
			// inserting actor-movie
			pstmt = conn.prepareStatement(
					"INSERT INTO actor_movie(actor_id, movie_id) VALUES(?, ?) " +
						"ON DUPLICATE KEY UPDATE actor_id=actor_id");
			
			int index = -1;		// starts from -1 because the first element index is 0
			for (Movie m : movieList) {
				index++;
				if (index >= actorList.size())
					break;
				Actor a = actorList.get(index);
			
				if (a.getId() == AbstPersistentEntity.newRecordID || m.getId() == AbstPersistentEntity.newRecordID)
					continue;
				if (a.isViolateRestrictions() || m.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear())) 
					continue;
				//check if connection already exist
				if (movieConnectionHash.containsKey(makeKeyFromMovieConnection(a.getId(), m.getId())))
					continue;
				
				// insert new actor-movie to batch
				pstmt.setInt(1, a.getId());	
				pstmt.setInt(2, m.getId());
				pstmt.addBatch();
				movieConnectionHash.put(makeKeyFromMovieConnection(a.getId(), m.getId()), true);
			}
			// executing the actor-movie batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			clearAllParams();
			throw new DB_Error("ERROR inserting batch actor-movie - " + e.getMessage());
		}
		
		// removing all actors and movies from list
		actorList.clear();
		movieList.clear();
		
		db.enqueueConnection(conn);
	}
		
	/**
	 * executing the batch insert/update director-movie
	 * @throws DB_Error if batch insert failed
	 */
	public void batchExecuteInsertDirectorMovie() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			ResultSet rs;
			Statement stmt = conn.createStatement();
			PreparedStatement pstmt = conn.prepareStatement(
					"INSERT INTO director(id, IMDB_first_name, IMDB_last_name, first_name, last_name) VALUES(?, ?, ?, ?, ?)");
			
			// calculating the current director id
			int currentDirectorId;
			rs = stmt.executeQuery("SELECT MAX(id) FROM director");
			if (rs.next()) 
				currentDirectorId = rs.getInt("MAX(id)") + 1;
			else
				currentDirectorId = 1;
			// closing the Statement
			stmt.close();
			
			// inserting directors
			for (Director d : directorList) {
				if (d.isViolateRestrictions()) 
					continue;
				if (d == previousDirector)
					continue;
				//check if director already exist
				if (directorsHash.containsKey(makeKeyFromDirector(d))) {
					d.setId(directorsHash.get(makeKeyFromDirector(d)));
					continue;
				}
				
				//insert new director to batch
				pstmt.setInt(1,currentDirectorId);	
				pstmt.setString(2, d.getFirstName());	
				pstmt.setString(3, d.getLastNameWithPersonNumber());
				pstmt.setString(4, d.getFirstName());	
				pstmt.setString(5, d.getLastName());
				pstmt.addBatch();
				
				previousDirector = d;
				d.setId(currentDirectorId);
				//insert director_key+id to hash
				directorsHash.put(makeKeyFromDirector(d), currentDirectorId);
				currentDirectorId++;
			}
			// executing the director batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
			// getting movies ids
			getMoviesIds();

			// inserting director-movie
			pstmt = conn.prepareStatement(
					"INSERT INTO director_movie(director_id, movie_id) VALUES(?, ?) " + 
						"ON DUPLICATE KEY UPDATE director_id=director_id");
			
			int index = -1;		// starts from -1 because the first element index is 0
			for (Movie m : movieList) {
				index++;
				if (index >= directorList.size())
					break;
				Director d = directorList.get(index);
				if (d.getId() == AbstPersistentEntity.newRecordID || m.getId() == AbstPersistentEntity.newRecordID)
					continue;
				if (d.isViolateRestrictions() || m.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear())) 
					continue;
				//check if connection already exist
				if (movieConnectionHash.containsKey(makeKeyFromMovieConnection(d.getId(), m.getId())))
					continue;
				
				// insert new director-movie to batch
				pstmt.setInt(1, d.getId());	
				pstmt.setInt(2, m.getId());
				pstmt.addBatch();
				movieConnectionHash.put(makeKeyFromMovieConnection(d.getId(), m.getId()), true);
			}
			// executing the director-movie batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			clearAllParams();
			throw new DB_Error("ERROR inserting batch director-movie - " + e.getMessage());
		}
		
		// removing all directors and movies from list
		directorList.clear();
		movieList.clear();
		
		db.enqueueConnection(conn);
	}
	
	/**
	 * executing the batch insert/update country-movie
	 * @throws DB_Error if batch insert failed
	 */
	public void batchExecuteInsertCountryMovie() throws DB_Error {
		Connection conn=db.dequeueConnection();

		try {
			ResultSet rs;
			Statement stmt = conn.createStatement();
			PreparedStatement pstmt = conn.prepareStatement(
					"INSERT INTO country(id, IMDB_name, name) VALUES(?, ?, ?)");
			
			// calculating the current country id
			int currentCountryId;
			rs = stmt.executeQuery("SELECT MAX(id) FROM country");
			if (rs.next()) 
				currentCountryId = rs.getInt("MAX(id)") + 1;
			else
				currentCountryId = 1;
			// closing the Statement
			stmt.close();
			
			// inserting countries
			for (Country c : countryList) {
				if (c.isViolateRestrictions()) 
					continue;
				//check if country already exist
				if (countriesHash.containsKey(c.getName().toLowerCase())) {
					c.setId(countriesHash.get(c.getName().toLowerCase()));
					continue;
				}
				
				//insert new country
				pstmt.setInt(1, currentCountryId);	
				pstmt.setString(2, c.getName());	
				pstmt.setString(3, c.getName());	
				pstmt.addBatch();
				
				//insert country_key+id to hash
				countriesHash.put(c.getName().toLowerCase(), currentCountryId);
				c.setId(currentCountryId);
				currentCountryId++;
			}
			// executing the country batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
			// getting movies ids
			getMoviesIds();

			// inserting country-movie
			pstmt = conn.prepareStatement(
					"INSERT INTO country_movie(country_id, movie_id) VALUES(?, ?) " +
						"ON DUPLICATE KEY UPDATE country_id=country_id");
			
			int index = -1;		// starts from -1 because the first element index is 0
			for (Movie m : movieList) {
				index++;
				if (index >= countryList.size())
					break;
				Country c = countryList.get(index);
				if (c.getId() == AbstPersistentEntity.newRecordID || m.getId() == AbstPersistentEntity.newRecordID)
					continue;
				if (c.isViolateRestrictions() || m.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear())) 
					continue;
				//check if connection already exist
				if (movieConnectionHash.containsKey(makeKeyFromMovieConnection(c.getId(), m.getId())))
					continue;
				
				// insert new country-movie to batch
				pstmt.setInt(1, c.getId());	
				pstmt.setInt(2, m.getId());
				pstmt.addBatch();
				movieConnectionHash.put(makeKeyFromMovieConnection(c.getId(), m.getId()), true);
			}
			// executing the country-movie batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			clearAllParams();
			throw new DB_Error("ERROR inserting batch country-movie - " + e.getMessage());
		}
		
		// removing all countries and movies from list
		countryList.clear();
		movieList.clear();
		
		db.enqueueConnection(conn);
	}
				
	/**
	 * executing the batch insert/update genre-movie
	 * @throws DB_Error if batch insert failed
	 */
	public void batchExecuteInsertGenreMovie() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			ResultSet rs;
			Statement stmt = conn.createStatement();
			PreparedStatement pstmt = conn.prepareStatement(
					"INSERT INTO genre(id, IMDB_name, name) VALUES(?, ?, ?)");
			
			// calculating the current genre id
			int currentGenreId;
			rs = stmt.executeQuery("SELECT MAX(id) FROM genre");
			if (rs.next()) 
				currentGenreId = rs.getInt("MAX(id)") + 1;
			else
				currentGenreId = 1;
			// closing the Statement
			stmt.close();
			
			// inserting genres
			for (Genre g : genreList) {
				if (g.isViolateRestrictions()) 
					continue;
				//check if country already exist
				if (genresHash.containsKey(g.getName().toLowerCase())) {
					g.setId(genresHash.get(g.getName().toLowerCase()));
					continue;
				}
				
				//insert new genre
				pstmt.setInt(1, currentGenreId);	
				pstmt.setString(2, g.getName());	
				pstmt.setString(3, g.getName());	
				pstmt.addBatch();
				
				//insert genre_key+id to hash
				genresHash.put(g.getName().toLowerCase(), currentGenreId);
				g.setId(currentGenreId);
				currentGenreId++;
			}
			// executing the country-movie batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();

			// getting movies ids
			getMoviesIds();

			// inserting genre-movie
			pstmt = conn.prepareStatement(
					"INSERT INTO genre_movie(genre_id, movie_id) VALUES(?, ?) " + 
						"ON DUPLICATE KEY UPDATE genre_id=genre_id");
			
			int index = -1;		// starts from -1 because the first element index is 0
			for (Movie m : movieList) {
				index++;
				if (index >= genreList.size())
					break;
				Genre g = genreList.get(index);
				if (g.getId() == AbstPersistentEntity.newRecordID || m.getId() == AbstPersistentEntity.newRecordID)
					continue;
				if (g.isViolateRestrictions() || m.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear())) 
					continue;
				//check if connection already exist
				if (movieConnectionHash.containsKey(makeKeyFromMovieConnection(g.getId(), m.getId())))
					continue;
				
				// insert new genre-movie to batch
				pstmt.setInt(1, g.getId());	
				pstmt.setInt(2, m.getId());
				pstmt.addBatch();
				movieConnectionHash.put(makeKeyFromMovieConnection(g.getId(), m.getId()), true);
			}
			// executing the genre-movie batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			clearAllParams();
			throw new DB_Error("ERROR inserting batch genre-movie - " + e.getMessage());
		}
		
		// removing all genres and movies from list
		genreList.clear();
		movieList.clear();
		
		db.enqueueConnection(conn);
	}
	
	/**
	 * executing the batch update ratings of movies
	 * @throws DB_Error if batch insert failed
	 */
	public void batchExecuteInsertRatingMovie() throws DB_Error {
		Connection conn=db.dequeueConnection();
		
		try {
			PreparedStatement pstmt = conn.prepareStatement(
					"UPDATE movie SET rank=?, votes=? WHERE id=?");
			
			// getting movies ids
			getMoviesIds();
			
			// updating movies ratings
			for (Movie m : movieList) {
				//check if movie id was found
				if (m.getId() == AbstPersistentEntity.newRecordID)
					continue;
				if (m.isViolateRestrictions(db.getMovieMinYear(), db.getMovieMaxYear())) 
					continue;
				//check if movie votes changed
				if (ratingsHash.containsKey(m.getId())) {
					if (ratingsHash.get(m.getId()) == m.getVotes())
						continue;
				}
				
				//insert update to batch
				pstmt.setFloat(1, m.getRank());	
				pstmt.setInt(2, m.getVotes());
				pstmt.setInt(3, m.getId());
				pstmt.addBatch();
			}
			// executing the ratings batch
			pstmt.executeBatch();
			// closing the Statement
			pstmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			clearAllParams();
			throw new DB_Error("ERROR inserting batch ratings of movies - " + e.getMessage());
		}
		
		// removing all movies from list
		movieList.clear();
		
		db.enqueueConnection(conn);
	}

	
	/**
	 * find the top movies and top actors in the DB
	 */
	public void findTopMoviesAndActors() throws DB_Error {
	Connection conn=db.dequeueConnection();
		
		try {
			Statement stmt = conn.createStatement();

			stmt.executeUpdate("UPDATE movie SET top_movie = 1 WHERE votes>="+topMovieVotes+" AND rank>"+topMovieRank);
			
			stmt.executeUpdate("UPDATE actor SET top_actor=1 WHERE id IN " +
									"(SELECT DISTINCT actor_id " +
									"FROM actor_movie x,top_movie " +
									"WHERE id=movie_id AND 10<=" +
										"(SELECT count(*) FROM actor_movie,top_movie WHERE id=movie_id AND actor_id=x.actor_id))");
			
			// closing the Statement
			stmt.close();
			
		} catch (SQLException e) {
			db.enqueueConnection(conn);
			clearAllParams();
			throw new DB_Error("ERROR finding top movies and actors - " + e.getMessage());
		}
		
		db.enqueueConnection(conn);
	}
	
	
	/**
	 * this function should be call after and import or an update 
	 */
	public void clearAllParams() {
		movieList.clear();
		actorList.clear();
		directorList.clear();
		countryList.clear();
		genreList.clear();
		
		moviesHash.clear();
		actorsHash.clear();
		directorsHash.clear();
		countriesHash.clear();
		genresHash.clear();
		ratingsHash.clear();
		
		movieConnectionHash.clear();
		
		previousActor = null;
		previousDirector = null;
		readHashFromDB = false;
		readConnectionHashFromDB = false;
	}
	
	
	//those functions are for debugging
	private void debugPrint(String s) {
		if (isDebug)
			System.out.println(s);
	}
	private String calculateElapsedTime(Boolean isStart) {
		if (! isDebug)
			return "";
		if (isStart) {
			start1 = new Date();
			return "";
		}
		else {
			if (start1 == null)
				return "";
			
			stop1 = new Date();
			
			long l1 = start1.getTime();
			long l2 = stop1.getTime();
			long diff = l2 - l1;
	
			long secondInMillis = 1000;
			long minuteInMillis = secondInMillis * 60;
			long hourInMillis = minuteInMillis * 60;
	
			long elapsedHours = diff / hourInMillis;
			diff = diff % hourInMillis;
			long elapsedMinutes = diff / minuteInMillis;
			diff = diff % minuteInMillis;
			long elapsedSeconds = diff / secondInMillis;
			
			start1 = null;
			return elapsedHours + ":" + elapsedMinutes + ":" + elapsedSeconds;
		}

	}
	private void calculateElapsedTime(Boolean isStart, String strToPrint) {
		if (! isDebug)
			return;
		DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
		Date date = new Date();
		
		if (isStart) {
			start = new Date();
			System.out.println("START:   " + strToPrint + "  at " + dateFormat.format(date) + ", counter=" + counter);
		}
		else {
			stop = new Date();
			
			long l1 = start.getTime();
			long l2 = stop.getTime();
			long diff = l2 - l1;
	
			long secondInMillis = 1000;
			long minuteInMillis = secondInMillis * 60;
			long hourInMillis = minuteInMillis * 60;
	
			long elapsedHours = diff / hourInMillis;
			diff = diff % hourInMillis;
			long elapsedMinutes = diff / minuteInMillis;
			diff = diff % minuteInMillis;
			long elapsedSeconds = diff / secondInMillis;
			
			System.out.println("FINSHED: " + strToPrint + "  at " + dateFormat.format(date) + ", counter=" + counter + ", " +
					"Time Elapsed=" + elapsedHours + ":" + elapsedMinutes + ":" + elapsedSeconds);
			System.out.println();
			counter = 0;
		}

	}

	
}
